package edu.buaa.act.model.tsc;

import com.graphhopper.util.GPXEntry;
import edu.buaa.act.AnonymousInterface;
import edu.buaa.act.anonymous.TrajectorySeparator;
import edu.buaa.act.helper.Config;
import edu.buaa.act.helper.RawDataCleaning;
import edu.buaa.act.helper.Utils;
import edu.buaa.act.model.GPSPoint;
import edu.buaa.act.model.Trajectory;
import edu.buaa.act.model.tsc.TrajectorySimpleConcat.EdgeNotFoundException;
import org.slf4j.Logger;

import java.io.*;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import static edu.buaa.act.helper.Arguments.Values.*;
import static edu.buaa.act.helper.Utils.toFix;

/**
 * Created by song on 17-7-4.
 */
public class ValidationTask extends Config
{
    private static Logger log = edu.buaa.act.helper.Utils.getDebugLogger();

    public static Config globalConf = new Config();


    public static List<ValidationTask> parseArgs(String[] args) throws IOException
    {
        log.trace("run with {} threads", args[8]);
        log.trace("dataPath:{}", args[0]);
        log.trace("nwaPath:{}", args[1]);
        log.trace("mapPath:{}", args[2]);
        log.trace("paramPath:{}", args[7]);
        log.trace("geoBound(S{} N{} W{} E{})", args[3], args[4], args[5], args[6]);

        globalConf.set(Raw_Traj_Data_File, args[0]);
        globalConf.set(Anonymity_NWA_EXE_Path, args[1]);
        globalConf.set(MapMatching_Map_Data_Path, args[2]);
        double south = Double.parseDouble(args[3]);
        double north = Double.parseDouble(args[4]);
        double west = Double.parseDouble(args[5]);
        double east = Double.parseDouble(args[6]);
        globalConf.set(Filter_Geo_Bound, new double[]{south, north, west, east});
        globalConf.set(Parallel_Count, Integer.parseInt(args[8]));

        List<ValidationTask> results = new ArrayList<>();

        BufferedReader reader = new BufferedReader(new FileReader(args[7]));
        String line=null;
        while((line=reader.readLine())!=null){
            if(!line.startsWith("#") && line.trim().length()>0){
                ValidationTask task = new ValidationTask();
                String[] arr = line.split(",");
                if(arr.length<15){
                    throw new RuntimeException("PARAMS ERROR: expect 15 params but got "+arr.length+" at line:\n"+line);
                }
                int i=0;
                task.set(Test_With_Anonymous_Data, Boolean.parseBoolean(arr[i].trim()));i++;
                task.set(Anonymous_Separately, Boolean.parseBoolean(arr[i].trim()));i++;
                task.set(Separate_Real_With_Time_Window, Boolean.parseBoolean(arr[i].trim()));i++;
                task.set(MapMatching_GPS_Accuracy, Integer.parseInt(arr[i].trim()));i++;

                task.set(Filter_Start_Time, Utils.rmQuote(arr[i]));i++;
                task.set(Filter_End_Time, Utils.rmQuote(arr[i]));i++;

                task.set(Test_Data_Count, Integer.parseInt(arr[i].trim())); i++;
                task.set(Test_Data_Filter_Start_Time, Utils.rmQuote(arr[i]));i++;
                task.set(Test_Data_Filter_End_Time, Utils.rmQuote(arr[i]));i++;
                task.set(Test_Data_L2S_Time_Interval, Integer.parseInt(arr[i].trim()));i++;

                task.set(Anonymity_Level_K, Integer.parseInt(arr[i].trim()));i++;
                task.set(Anonymity_Interpolation_Time_Interval, Integer.parseInt(arr[i].trim()));i++;

                task.set(Anonymity_L2S_Time_Window_Size, Integer.parseInt(arr[i].trim()));i++;
                task.set(Anonymity_L2S_Time_Window_Start_Threshold, Integer.parseInt(arr[i].trim()));i++;
                task.set(Anonymity_L2S_Time_Window_End_Threshold, Integer.parseInt(arr[i].trim()));i++;

                results.add(task);
            }
        }

        log.trace("read {} tasks. global configure: {}", results.size(),
                globalConf.toString(
                        Raw_Traj_Data_File,
                        Anonymity_NWA_EXE_Path,
                        MapMatching_Map_Data_Path,
                        Filter_Geo_Bound,
                        Parallel_Count
                ));
        return results;
    }


    public void run(List<List<GPSPoint>> data) throws ParseException, IOException, InterruptedException
    {
        Thread.currentThread().setName("task");

        String gStart = this.get(Filter_Start_Time);
        String gEnd = this.get(Filter_End_Time);

        data = RawDataCleaning.filterByTime(data, gStart, gEnd);

        if(this.get(Test_With_Anonymous_Data))
        {
            if(this.get(Anonymous_Separately))
            {
                this.testSeparateAnonymousData(data);
            }else{
                this.testWithAnonymousData(data);
            }
        }else{
            this.testWithRealData(data);
        }
    }

    public void testWithRealData(List<List<GPSPoint>> data) throws ParseException, IOException, InterruptedException
    {
        String start = this.get(Test_Data_Filter_Start_Time);
        String end   = this.get(Test_Data_Filter_End_Time);
        int sampleCount = this.get(Test_Data_Count);

        AnonymousSampler sampler = new AnonymousSampler(data, sampleCount, start, end);

        int winSize = this.get(Anonymity_L2S_Time_Window_Size);
        int winStart = this.get(Anonymity_L2S_Time_Window_Start_Threshold);
        int winEnd = this.get(Anonymity_L2S_Time_Window_End_Threshold);

        int k = this.get(Anonymity_Level_K);
        List<List<GPSPoint>> trainData = this.anonymous(sampler.getTrainingData(), k);

        sunshe.CallBack.afterAnonymous(trainData, this);

//        return;

        List<List<GPSPoint>> randomTrainingData = sunshe.CallBack.getRandomData();
        if(randomTrainingData!=null) {
            // test by random and real training data
            List<List<Trajectory>> trainTrajs = TrajectorySeparator.smartSeparate(trainData, winSize, winStart, winEnd);
            List<List<Trajectory>> trainRandom = TrajectorySeparator.smartSeparate(randomTrainingData, winSize, winStart, winEnd);
            List<List<Trajectory>> testTrajs = TrajectorySeparator.smartSeparate(sampler.getTestData(), winSize, winStart, winEnd);

            SyncModel modelReal = new SyncModel(this);
            modelReal.start();
            modelReal.feedTest(testTrajs);
            modelReal.feedTrain(trainTrajs);
            modelReal.join();

            SyncModel modelRandom = new SyncModel(this);
            modelRandom.start();
            modelRandom.feedTest(testTrajs);
            modelRandom.feedTrain(trainRandom);
            modelRandom.join();
        }else {
            Model model = new Model(this, sampler);
            model.start();

            trainData = sunshe.CallBack.addRandomNoise(trainData);

            List<List<Trajectory>> trainTrajs = TrajectorySeparator.smartSeparate(trainData, winSize, winStart, winEnd);

            sunshe.CallBack.afterSeparation(trainTrajs, this);

            model.feedTrain(trainTrajs);
            model.join();
        }
    }


    public void testWithAnonymousData(List<List<GPSPoint>> data) throws ParseException, IOException, InterruptedException
    {
        int k = this.get(Anonymity_Level_K);
        data = this.anonymous(data, k);

        String start = this.get(Test_Data_Filter_Start_Time);
        String end   = this.get(Test_Data_Filter_End_Time);
        int sampleCount = this.get(Test_Data_Count);

        AnonymousSampler sampler = new AnonymousSampler(data, sampleCount, start, end);


        SyncModel model = new SyncModel(this);
        model.start();


        int winSize = this.get(Anonymity_L2S_Time_Window_Size);
        int winStart = this.get(Anonymity_L2S_Time_Window_Start_Threshold);
        int winEnd = this.get(Anonymity_L2S_Time_Window_End_Threshold);
        List<List<Trajectory>> trainTrajs = TrajectorySeparator.smartSeparate(sampler.getTrainingData(), winSize, winStart, winEnd);
        List<List<Trajectory>> testTrajs = TrajectorySeparator.smartSeparate(sampler.getTestData(), winSize, winStart, winEnd);

        model.feedTest(testTrajs);
        model.feedTrain(trainTrajs);

        model.join();

    }

    public void testSeparateAnonymousData(List<List<GPSPoint>> data) throws ParseException, IOException, InterruptedException
    {
        String start = this.get(Test_Data_Filter_Start_Time);
        String end   = this.get(Test_Data_Filter_End_Time);
        int sampleCount = this.get(Test_Data_Count);
        int k = this.get(Anonymity_Level_K);

        AnonymousSampler sampler = new AnonymousSampler(data, sampleCount, start, end);
        List<List<GPSPoint>> testData = new ArrayList<>();
        for(List<GPSPoint> traj : sampler.getTestData()){

            if(traj.size()>10){
                long tStart = traj.get(0).getTime();
                long tEnd = traj.get(traj.size()-1).getTime();
                if(tEnd-tStart>(sampler.getSampleEndTimestamp()-sampler.getSampleStartTimestamp())*0.5)//prevent no overlap err, but maybe too strict.
                {
                    testData.add(traj);
                }
            }
        }
        testData = this.anonymous(testData, k);
        List<List<GPSPoint>> trainData = this.anonymous(sampler.getTrainingData(), k);


        int winSize = this.get(Anonymity_L2S_Time_Window_Size);
        int winStart = this.get(Anonymity_L2S_Time_Window_Start_Threshold);
        int winEnd = this.get(Anonymity_L2S_Time_Window_End_Threshold);
        List<List<Trajectory>> testShortTrajs = TrajectorySeparator.smartSeparate(testData, winSize, winStart, winEnd);
        List<List<Trajectory>> trainShortTrajs = TrajectorySeparator.smartSeparate(trainData, winSize, winStart, winEnd);

        SyncModel model = new SyncModel(this);
        model.start();
        model.feedTest(testShortTrajs);
        model.feedTrain(trainShortTrajs);
        model.join();
    }


    private List<List<GPSPoint>> anonymous(List<List<GPSPoint>> data, int k) throws IOException
    {
        if(k<=0){
            throw new RuntimeException("anonymous k must be positive");
        }else if(k>data.size())
        {
            throw new RuntimeException("K should smaller than traj count, but K="+k+">"+data.size());
        }else if(k==1){
            return data;
        }else{
            int interval = this.get(Anonymity_Interpolation_Time_Interval);
//            return AnonymousInterface.anonymousByExternalCProgram(data, k, interval);
            return AnonymousInterface.anonymousByJava(data, k, interval);
        }
    }




    private static class Model extends Thread
    {
        protected TrajectorySimpleConcat tscModel = new TrajectorySimpleConcat();
        private Config config;
        protected MapMatchingServer testServer;
        protected MapMatchingServer trainServer;
        private LinkedBlockingQueue<Trajectory> in = new LinkedBlockingQueue<>();
        private AnonymousSampler testDataSampler;
        private volatile boolean canExit = false;

        Model(ValidationTask validationTask, AnonymousSampler sampler) throws IOException {
            this(validationTask);

            this.testDataSampler = sampler;
        }

        private Model(ValidationTask validationTask) throws IOException
        {
            this.config = validationTask;

            this.testServer = newMMServer(true);
            this.testServer.setOutput(in);

            this.trainServer = newMMServer(true);
            this.trainServer.setOutput(in);
        }

        public void run()
        {
            Thread.currentThread().setName("model");

            feedTest();

            buildModelAndValidate();
        }

        protected void buildModelAndValidate()
        {
            try {
                List<Trajectory> samples = buildModel();

                this.testServer.join();
                this.trainServer.join();

                sunshe.CallBack.mmFailedTrajs(trainServer.getFailedTrajs(), testServer.getFailedTrajs(), config);

                List<Trajectory> realSamples = new ArrayList<>();

                for (Trajectory traj : samples)
                {
                    try {
                        tscModel.validateExternalTraj(traj);
                        realSamples.add(traj);
                    } catch (EdgeNotFoundException ignore) {

                    }
                }

                this.calcResult(realSamples);

                log.debug("merged road count in TSC model: {}", tscModel.getMergedRoadCount());

                config.set(Train_Data_History_Road_Count, tscModel.getHistoryRoadCount());
                config.set(Train_Data_Recent_Road_Count, tscModel.getTodayRoadCount());

                config.set(Test_Data_MM_IN_OUT_RATIO, toFix(testServer.getInOutRate(),3));
                config.set(Train_Data_MM_IN_OUT_RATIO, toFix(trainServer.getInOutRate(), 3));

                log.trace("real test with {} short trajectories, {} ignored", realSamples.size(), samples.size()-realSamples.size());

                log.debug("\nRESULTS:\n{}", config.toString(false));

                log.info("RESULTS(in one line):\n{}", config.toString(true));

                int test = config.get(Test_Data_Traj_Count);
                long trainH = config.get(Train_Data_History_Traj_Count);
                long trainR = config.get(Train_Data_Recent_Traj_Count);

                log.debug("test ratio({}/{}+{})={}%", test, test, trainH+trainR, Utils.percent(test, test+trainH+trainR));

//                log.info("\n## Important Output Results\n{}", config.toTableString(
//                        Mean_Absolute_Error, Mean_Relative_Error, MAE_Per_Kilometer,
//                        Train_Traj_AVG_Length, Train_Traj_AVG_Time, Train_Data_MM_IN_OUT_RATIO,
//                        Test_Traj_AVG_Length, Test_Traj_AVG_Time, Test_Data_MM_IN_OUT_RATIO
//                ));

                logImportantHumanReadableResult();


            } catch (InterruptedException e) {
                log.warn("model thread interrupt", e);
            }
        }

        private void logImportantHumanReadableResult()
        {
            log.debug("MAE({}/{}#) MRE({}/{}hour) MAEL({}/{}km)",
                    toFix(config.<Float>get(Mean_Absolute_Error),2),
                    config.<Integer>get(Test_Data_Traj_Count),
                    toFix(config.<Float>get(Mean_Relative_Error),3),
                    toFix(config.<Integer>get(Test_Data_Total_Time)/3600f, 1),
                    toFix(config.<Float>get(MAE_Per_Kilometer), 2),
                    toFix(config.<Integer>get(Test_Data_Total_Length)/1000f, 1)
            );

            int testL = config.get(Test_Data_Total_Length);
            int testT = config.get(Test_Data_Total_Time);
            long testDL = config.get(Test_Data_Direct_Length);

            long trainLH = config.get(Train_Data_History_Total_Length);
            long trainLR = config.get(Train_Data_Recent_Total_Length);
            long trainTH = config.get(Train_Data_History_Total_Time);
            long trainTR = config.get(Train_Data_Recent_Total_Time);
            long trainDL = config.get(Train_Data_Direct_Length);

            log.debug("Compare:          Train vs Test");
            log.debug("MM SUCCESS RATIO   {}% | {}%", config.<Double>get(Train_Data_MM_IN_OUT_RATIO)*100, config.<Double>get(Test_Data_MM_IN_OUT_RATIO)*100);
            log.debug("MM LENGTH RATIO     {} | {}", toFix((trainLH+trainLR)*1d/trainDL,2), toFix(testL*1d/testDL, 2));
            log.debug("AVG LENGTH        {}km | {}km", config.<Long>get(Train_Traj_AVG_Length)/1000, config.<Long>get(Test_Traj_AVG_Length)/1000);
            log.debug("AVG TIME         {}min | {}min", config.<Long>get(Train_Traj_AVG_Time)/60,    config.<Long>get(Test_Traj_AVG_Time)/60);
            log.debug("AVG SPEED       {}km/h | {}km/h", toFix((trainLH+trainLR)/1000f/((trainTH+trainTR)/3600f),2), toFix(testL/1000f/(testT/3600f),2));
            log.debug("NUMBER           {} | {}",  config.<Long>get(Train_Data_Recent_Traj_Count)+config.<Long>get(Train_Data_History_Traj_Count),config.<Integer>get(Test_Data_Traj_Count));
        }

        private void feedTest()
        {
            boolean l2sTimeWin = config.get(Separate_Real_With_Time_Window);

            Thread t = new Thread(() -> {
                List<List<GPSPoint>> test = testDataSampler.getTestData();
                List<List<Trajectory>> after;
                if(l2sTimeWin)
                {
                    int winSize = config.get(Anonymity_L2S_Time_Window_Size);
                    int winStart = config.get(Anonymity_L2S_Time_Window_Start_Threshold);
                    int winEnd = config.get(Anonymity_L2S_Time_Window_End_Threshold);
                    after = TrajectorySeparator.smartSeparate(test, winSize, winStart, winEnd);
                }
                else
                {
                    int interval = config.get(Test_Data_L2S_Time_Interval);
                    after = TrajectorySeparator.separate(test, interval);
                }
                try
                {
                    testServer.start();

                    for(List<Trajectory> lst : after)
                    {
                        for (Trajectory traj : lst)
                        {
                            traj.setSample(true);
                            testServer.put(traj);
                        }
                    }
                    log.trace("test data fed.");
                } catch (InterruptedException e) {
                    log.warn("feeding test interrupted.",e);
                }
            });
            t.start();
        }

        private List<Trajectory> buildModel() throws InterruptedException
        {
            List<Trajectory> samples = new ArrayList<>();
            long trainHistoryCount = 0;
            long trainHistoryLength = 0;
            long trainHistoryTime = 0;
            long trainTodayCount = 0;
            long trainTodayLength = 0;
            long trainTodayTime = 0;
            long trainDirectLength = 0;
            int waitCount=0;
            List<Trajectory> trainingData = new ArrayList<>();
            while(true)
            {
                Trajectory traj = in.poll(1, TimeUnit.SECONDS);
                if(traj==null){
                    if(this.canExit && this.testServer.done() && this.trainServer.done()){
                        break;
                    }else{
                        waitCount++;
                        if(waitCount==100){
                            this.testServer.printStatus();
                            this.trainServer.printStatus();
                            waitCount=0;
                        }
                    }
                }else {
                    if(traj.isSample())
                    {
                        samples.add(traj);
                    }else{
                        trainingData.add(traj);
                        if (traj.isHistory())
                        {
                            tscModel.readHistoryTrajInc(traj);
                            trainHistoryCount++;
                            trainHistoryLength+=traj.getLength();
                            trainHistoryTime += traj.realTravelTime();
                            trainDirectLength += traj.getDirectLength();
                        } else {
                            tscModel.readTrajInc(traj);
                            trainTodayCount++;
                            trainTodayLength += traj.getLength();
                            trainTodayTime += traj.realTravelTime();
                            trainDirectLength += traj.getDirectLength();
                        }
                    }
                }
            }
            sunshe.CallBack.afterMapMatching(samples, trainingData, config);

            this.trainServer.showLog();
            this.testServer.showLog();

            config.set(Train_Data_History_Traj_Count, trainHistoryCount);
            config.set(Train_Data_History_Total_Length, trainHistoryLength);
            config.set(Train_Data_History_Total_Time, trainHistoryTime);

            config.set(Train_Data_Recent_Traj_Count, trainTodayCount);
            config.set(Train_Data_Recent_Total_Length, trainTodayLength);
            config.set(Train_Data_Recent_Total_Time, trainTodayTime);

            config.set(Train_Data_Direct_Length, trainDirectLength);

            long trainCount = (trainHistoryCount+trainTodayCount);
            if(trainCount>0) {
                long avgLen = (trainHistoryLength + trainTodayLength) / trainCount;
                long avgTime = (trainHistoryTime + trainTodayTime) / trainCount;
                config.set(Train_Traj_AVG_Length, avgLen);
                config.set(Train_Traj_AVG_Time, avgTime);
            }else{
                config.set(Train_Traj_AVG_Length, 0);
                config.set(Train_Traj_AVG_Time, 0);
            }
            log.trace("model built.");

            return samples;
        }

        private MapMatchingServer newMMServer(boolean allowCut) throws IOException {
            String mapPath = globalConf.get(MapMatching_Map_Data_Path);
            int parallelCount = globalConf.get(Parallel_Count);
            int gpsAccuracy = config.get(MapMatching_GPS_Accuracy);

            return new MapMatchingServer(parallelCount, Utils.getExistFile(mapPath), gpsAccuracy, allowCut);
        }

        public void feedTrain(List<List<Trajectory>> trainTrajs) throws ParseException, InterruptedException {
            String sampleStartTime = config.get(Test_Data_Filter_Start_Time);
            SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            long todayHistorySeparator = df.parse(sampleStartTime).getTime();

            this.trainServer.start();
            for(List<Trajectory> lst : trainTrajs){
                for(Trajectory traj : lst)
                {
                    traj.setSample(false);

                    List<GPXEntry> gps = traj.getGpsEntries();
                    if(!gps.isEmpty()){
                        if(gps.get(0).getTime()>todayHistorySeparator) {
                            traj.setHistory(false);
                            this.trainServer.put(traj);
                        }else if(gps.get(gps.size()-1).getTime()<todayHistorySeparator){
                            traj.setHistory(true);
                            this.trainServer.put(traj);
                        }else{
                            // do nothing.
                        }
                    }
                }
            }

            this.canExit = true;
        }

        private void calcResult(List<Trajectory> samples)
        {
            sunshe.CallBack.afterModel(samples, config);
            if(samples.size()>0) {
                double sumOfErr = 0;
                double sumOfTruth=0;
                double totalLength=0;
                double directLength=0;
                for(Trajectory t : samples){ // for every sample
                    directLength += t.getDirectLength();
                    sumOfErr += Math.abs(t.getEstimateTravelTime() - t.realTravelTime());
                    sumOfTruth += t.realTravelTime();
                    totalLength += t.getLength();
                }
                config.set(Mean_Absolute_Error, (float) (sumOfErr / samples.size() / 60)); // second to minute
                config.set(Mean_Relative_Error, (float) (sumOfErr / sumOfTruth ));
                config.set(MAE_Per_Kilometer, (float) (sumOfErr / 60 / (totalLength / 1000))); // m to km
                config.set(Test_Data_Traj_Count, samples.size());
                config.set(Test_Data_Total_Length, (int) totalLength);
                config.set(Test_Data_Total_Time, (int) sumOfTruth);
                config.set(Test_Traj_AVG_Length, (long) (totalLength/samples.size()));
                config.set(Test_Traj_AVG_Time, (long) (sumOfTruth/samples.size()));
                config.set(Test_Data_Direct_Length, (long) directLength);
            }else{
                config.set(Mean_Absolute_Error, Double.NaN);
                config.set(Mean_Relative_Error, Double.NaN);
                config.set(MAE_Per_Kilometer,  Double.NaN);
                config.set(Test_Data_Traj_Count,   0);
                config.set(Test_Data_Total_Length,  0);
                config.set(Test_Data_Total_Time, 0);
                config.set(Test_Traj_AVG_Length, 0);
                config.set(Test_Traj_AVG_Time, 0);
            }
        }

    }

    private static class SyncModel extends Model
    {
        SyncModel(ValidationTask validationTask) throws IOException, InterruptedException {
            super(validationTask);
        }

        public void run()
        {
            Thread.currentThread().setName("model");

            buildModelAndValidate();
        }

        public void feedTest(List<List<Trajectory>> test) {
            try
            {
                testServer.start();

                for(List<Trajectory> lst : test)
                {
                    for (Trajectory traj : lst)
                    {
                        traj.setSample(true);
                        testServer.put(traj);
                    }
                }
                log.trace("test data fed.");
            } catch (InterruptedException e) {
                log.warn("feeding test interrupted.",e);
            }
        }
    }


}
